use std::collections::HashMap;

use datafusion::arrow::datatypes::{DataType, Field, Schema};

use crate::{Error, LabelType, LResult};

/// SchemaMut
/// ts_s ts seconds, u64
/// ts_n ts nano seconds, u32
/// $t_xx all tags are suffix with $t_
/// $f_xx all fields are suffix with $f_
pub struct SchemaMut {
    pub fields: Vec<Field>,
    pub metadata: HashMap<String, String>,
}

impl SchemaMut {
    pub fn create() -> Self {
        let ts_s = Field::new("ts_s", DataType::UInt64, false);
        let ts_n = Field::new("ts_n", DataType::UInt32, false);
        Self {
            fields: vec![ts_s, ts_n],
            metadata: Default::default(),
        }
    }

    pub fn ensure_tag(&mut self, tag: &str) {
        let real_tag_name = Self::real_tag_name(tag);
        if !self.contains_real_field(&real_tag_name) {
            self.fields.push(Field::new(real_tag_name, DataType::UInt64, true));
        }
    }

    pub fn ensure_field(&mut self, field: &str, data_type: DataType) {
        let real_field_name = Self::real_field_name(field);
        if !self.contains_real_field(&real_field_name) {
            self.fields.push(Field::new(real_field_name, data_type, true));
        }
    }

    pub fn get_tag(&self, tag: &str) -> Option<&Field> {
        let real_tag_name = Self::real_tag_name(tag);
        self.fields.iter().find(|f| f.name().eq(&real_tag_name))
    }

    pub fn get_field(&self, tag: &str) -> Option<&Field> {
        let real_tag_name = Self::real_field_name(tag);
        self.fields.iter().find(|f| f.name().eq(&real_tag_name))
    }

    #[inline]
    pub fn contains_real_field<S: AsRef<str>>(&self, name: S) -> bool {
        let s = name.as_ref();
        self.fields.iter().any(|f| f.name().eq(s))
    }

    #[inline]
    pub fn is_real_tag(name: &str) -> bool {
        name.starts_with("$t_")
    }

    #[inline]
    fn real_tag_name(tag_name: &str) -> String {
        format!("$t_{}", tag_name)
    }
    #[inline]
    fn real_field_name(field_name: &str) -> String {
        format!("$f_{}", field_name)
    }

    pub fn dismiss_name(s: &String) -> LResult<(&str, LabelType)> {
        let sub = &s.as_bytes()[3..];
        let name = unsafe {
            std::str::from_utf8_unchecked(sub)
        };
        if s.starts_with("$t_") {
            Ok((name, LabelType::Tag))
        } else if s.starts_with("$f_") {
            Ok((name, LabelType::Field))
        } else {
            return Err(Error::StorageErr(format!("MemTable illegal label: {}", s)));
        }
    }

    pub fn tag_index(&self, real_tag_name: &String) -> LResult<usize> {
        let mut index = 0;
        for f in &self.fields[2..] {
            if f.name().starts_with("$t_") {
                if f.name().eq(real_tag_name) {
                    return Ok(index);
                }
                index += 1;
            }
        }
        Err(Error::StorageErr(format!(
            "SchemaMut#tag_index real_tag_name: {} not exists", &real_tag_name
        )))
    }

    #[inline]
    pub fn as_schema(&self) -> Schema {
        Schema::new_with_metadata(self.fields.clone(), self.metadata.clone())
    }
}